package com.liu.kafka.listener;

import cn.hutool.core.thread.ThreadUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.UUID;

/**
 * 批量消费
 */
@Slf4j
@Component
public class BatchConsumer {


    public static final String KAFKA_LISTENER_ID = "first";

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

//    @PostConstruct
    public void producer() {
        ThreadUtil.sleep(5000);
        new Thread(() -> {
            while (true) {
                String msg = UUID.randomUUID().toString();
                //kafka声明式事务
                kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<String, String, Object>() {
                    @Override
                    public Object doInOperations(KafkaOperations<String, String> operations) {
                        operations.send(KAFKA_LISTENER_ID, msg);
                        return null;
                    }
                });
                ThreadUtil.sleep(500);
            }
        }).start();
    }

    /**
     * 批量拉取,同步多线程处理数据
     *
     * @param records
     */
    @KafkaListener(topics = KAFKA_LISTENER_ID, groupId = "test1", containerFactory = "batchFactory")
    public void batchPullMessage(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        try {
            log.info("-----------------kafka拉取的消息数量: {}", records.size());
            for (ConsumerRecord<String, String> record : records) {
                log.info("value: {}", record.value());
            }
            ack.acknowledge();//提交kafka的offset
            ThreadUtil.sleep(1000);
        } catch (Exception e) {
            log.error("kafka消费者异常", e);
        }
    }

}

